package com.jaymin.flink.datastream.source;

import com.google.common.collect.Lists;
import com.jaymin.flink.datastream.transformation.Access;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * @author jaymin
 * @since 2024/4/5 19:13
 */
public class AccessSource implements SourceFunction<Access> {
    protected boolean running = true;

    @Override
    public void run(SourceContext<Access> ctx) throws Exception {
        List<String> domains = Lists.newArrayList("com.immocc", "com.b", "com.a");
        Random random = new Random();
        while (running) {
            for (int i = 0; i < 10; i++) {
                Access element = new Access();
                element.setTime(123456L);
                element.setDomain(domains.get(random.nextInt(domains.size())));
                element.setTraffic(new BigDecimal(random.nextDouble()));
                ctx.collect(element);
            }
            Thread.sleep(5000);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
